package com.yinhai.uid.worker;

import com.yinhai.uid.utils.WorkIdUtils;
import com.yinhai.uid.utils.jerseyUtils;
import com.yinhai.uid.zookeeper.ZookeeperRegistryCenter;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.File;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

public class ZookeeperWorkerIdAssigner implements WorkerIdAssigner {

	@Autowired
	private ZookeeperRegistryCenter zookeeperRegistryCenter;

	private String zkNodeRegistryName;

	private String workerIdFileBasePath;

	private long launchTimeOffset;

	private static final String FOREVER_PATH = "/forever";

	private static final String TEMPORARY_PATH = "/temporary";

	private static final String FILE_PATH_SEPARATOR = File.separator;

	private static final String ZK_PATH_SEPARATOR = "/";

	private static final String HTTP_PREFIX = "http://";

	private static final String SYSTEM_TIME_URL_SUFFIX = "/uid/getSysTime";

	@Override
	public long assignWorkerId() {
		if (!zookeeperRegistryCenter.isExisted(FOREVER_PATH)) {
			zookeeperRegistryCenter.createPersistent(FOREVER_PATH);
		}
		StringBuilder sb = new StringBuilder();
		sb.append(FOREVER_PATH).append(ZK_PATH_SEPARATOR);
		List<String> registryNodes = zookeeperRegistryCenter.getChildrenKeys(FOREVER_PATH);
		boolean isRegistry = false;
		if (registryNodes != null && registryNodes.size() > 0) {
			for (String node : registryNodes) {
				if (zkNodeRegistryName.equals(node.substring(0, node.length() - 10))) {
					sb.append(node);
					isRegistry = true;
					break;
				}
			}
		}
		if (!isRegistry) {
			sb.append(zkNodeRegistryName);
		}
		String serverPath = sb.toString();
		long workerId = -1L;
		boolean isCacheFile = false;
		// 已经写过此节点
		if (isRegistry) {
			// 判断机器时间是否发生了大步长回拨
			long recordTime = Long.parseLong(zookeeperRegistryCenter.get(serverPath));
			long sysTime = System.currentTimeMillis();
			if (sysTime < recordTime) {
				throw new RuntimeException("机器时间发生了大步长回拨，启动失败");
			}
		} else {
			// 未写过此节点，直接创建持久节点'/forever/${self}'并写入自身系统时间
			long selfSysTime = System.currentTimeMillis();
			zookeeperRegistryCenter.createPersistentSequential(serverPath,Long.toString(selfSysTime));
           // 综合对比其余节点的系统时间来判断自身系统时间是否准确
			List<String> tempNodes = zookeeperRegistryCenter.getChildrenKeys(TEMPORARY_PATH);
			long sum = 0L;
			for (String path : tempNodes) {
				sb.delete(0,sb.length());
				sb.append(HTTP_PREFIX).append(path).append(SYSTEM_TIME_URL_SUFFIX);
				Object result = jerseyUtils.sendGet(sb.toString());
				if (result != null) {
					long sysTime = (Long) result;
					sum += sysTime;
				}
			}
			if (tempNodes != null && tempNodes.size() >  0) {
				long averageTime = sum / tempNodes.size();
				// 若abs( 系统时间-sum(time)/nodeSize ) < 阈值，抛异常
				if (Math.abs(selfSysTime - averageTime) >= launchTimeOffset) {
					throw new RuntimeException("本机系统时间发生大步长偏移，启动失败");
				}
			}
			isCacheFile = true;
		}
		// 写临时节点/temporary/${self}
		zookeeperRegistryCenter.createEphemeral(TEMPORARY_PATH + ZK_PATH_SEPARATOR + zkNodeRegistryName);
		// 根据子节点数量获取workId
		List<String> workerIdNodes = zookeeperRegistryCenter.getChildrenKeys(FOREVER_PATH);
		// 根据zk顺序节点生成的序列号排序
		Collections.sort(workerIdNodes, new Comparator<String>() {
			@Override
			public int compare(final String o1, final String o2) {
				return o1.substring(o1.length()-10,o1.length()).compareTo(o2.substring(o2.length()-10,o2.length()));
			}
		});
		// 当前集群节点在集合中的顺序号就是workerId，保证了workerId是递增的，不会重复
		for (int i = 0; i < workerIdNodes.size(); i++) {
			if (workerIdNodes.get(i).startsWith(zkNodeRegistryName)) {
				workerId = i + 1;
				break;
			}
		}
		// 缓存workerID文件到本地文件系统，文件路径格式：[workerIdFileBasePath + /zkNodeRegistryName + /workerId] (如：/workerId/192.168.17.31-8080/1)
		String directory = zkNodeRegistryName.replace(":","-");
		if (isCacheFile) {
			sb.delete(0,sb.length());
			sb.append(workerIdFileBasePath)
					.append(FILE_PATH_SEPARATOR)
					.append(directory)
					.append(FILE_PATH_SEPARATOR)
					.append(workerId);
			WorkIdUtils.writeWorkIdFile(sb.toString());
		}
		// zookeeper出现问题,从本机硬盘获取workId
		if (workerId == -1) {
			sb.delete(0,sb.length());
			sb.append(workerIdFileBasePath)
					.append(FILE_PATH_SEPARATOR)
					.append(directory);
			workerId = WorkIdUtils.getWorkIdFromDisk(sb.toString());
		}
		return workerId;
	}

	public void setZkNodeRegistryName(String zkNodeRegistryName) {
		this.zkNodeRegistryName = zkNodeRegistryName;
	}

	public void setWorkIdFileBasePath(String workerIdFileBasePath) {
		this.workerIdFileBasePath = workerIdFileBasePath;
	}

	public void setLaunchTimeOffset(long launchTimeOffset) {
		this.launchTimeOffset = launchTimeOffset;
	}
}
